package com.wsq.rxjavademo.resp;

import android.util.Log;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

/**
 * Date:2022/11/24
 * Time:21:44
 * author:wushengqi
 */
public class CacheRes {

    private String TAG = CacheRes.class.getSimpleName();

    private List<Integer> cacheData = new ArrayList<>();

    private static volatile CacheRes cacheRes;

    private AtomicBoolean updating = new AtomicBoolean(false);

    private BlockingQueue<CacheCallback> cacheCallbacks = new LinkedBlockingQueue<>();
    private BlockingQueue<Integer> waitingUpdateData = new LinkedBlockingQueue<>();

    //更新本地数据，均应加锁处理
    private Object lock = new Object();

    private CacheRes(){
        init();
    }

    public static CacheRes getIntance(){
        if (cacheRes == null){
            synchronized (CacheRes.class){
                if (cacheRes == null){
                    cacheRes = new CacheRes();
                }
            }
        }

        return cacheRes;
    }

    private void init(){
        if (cacheData.size() == 0){
            refreshData(null);
        }
    }

    public void updateInteger(Integer integer){
        if (integer == null){
            return;
        }

        if (updating.compareAndSet(true, true)){
            try {
                waitingUpdateData.put(integer);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return;
        }

        synchronized (lock){

        }
    }

    public void getCache(CacheCallback cacheCallback){

        if (cacheCallback == null){
            return;
        }

        //如果当前值为true
        if (updating.compareAndSet(true, true)){
            addCallback(cacheCallback);
            return;
        }
        //获取缓存数据的地方也做加锁处理，避免读取数据的时候，有写数据的操作
        synchronized (lock){
            cacheCallback.cacheCallbackFromMainThread(deepCopy(cacheData));
        }
    }

    public void refreshData(CacheCallback cacheCallback){
        if (cacheCallback == null){
            return;
        }

        if (updating.compareAndSet(true, true)){
            addCallback(cacheCallback);
        }
        mockNetRequest(cacheCallback);
    }

    private void mockNetRequest(CacheCallback cacheCallback){
        if (cacheCallback == null){
            return;
        }

        //如果预期是false，则更新为true
        if (updating.compareAndSet(false, true)){
            Observable.create(new ObservableOnSubscribe<List<Integer>>() {
                @Override
                public void subscribe(ObservableEmitter observableEmitter) throws Exception {

                    Log.e(TAG, "网络请求发生在线程：" + Thread.currentThread().getName());
                    List<Integer>  list = new ArrayList<>();
                    for (int i = 0; i < new Random().nextInt() + 10000; i++){
                        list.add(i);
                    }
                    Log.e(TAG, "mockNetRequest:  length=" + list.size());

                    observableEmitter.onNext(list);

                    Thread.sleep(3000);
                    observableEmitter.onComplete();
                }
            }).subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Observer<List<Integer>>() {
                        @Override
                        public void onSubscribe(Disposable disposable) {
                            Log.e(TAG, "onSubscribe");
                        }

                        @Override
                        public void onNext(List<Integer> integers) {
                            Log.e(TAG, "回调结束发生在线程：" + Thread.currentThread().getName());
                            Log.e(TAG, "onNext");
                            synchronized (lock){
                                cacheData.clear();
                                cacheData = integers;

                                updating.compareAndSet(true, false);
                            }


                        }

                        @Override
                        public void onError(Throwable throwable) {
                            Log.e(TAG, "onError");
                        }

                        @Override
                        public void onComplete() {
                            Log.e(TAG, "onComplete");
                        }
                    });
        }else{
            addCallback(cacheCallback);
        }


    }

    public interface CacheCallback{
        void cacheCallbackFromMainThread(List<Integer> integers);
    }

    private void addCallback(CacheCallback cacheCallback){
        if (!cacheCallbacks.contains(cacheCallback)){
            cacheCallbacks.add(cacheCallback);
        }
    }

    /**
     * 调用缓存的回调方法，将数据返回回去
     * @param datas 数据
     */
    private void callbackDatas(List<Integer> datas){
        while (!cacheCallbacks.isEmpty()){
            try {
                CacheCallback cacheCallback = cacheCallbacks.take();
                cacheCallback.cacheCallbackFromMainThread(deepCopy(datas));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    //对数据做深拷贝
    private List<Integer> deepCopy(List<Integer> datas){
        List<Integer> data = new ArrayList<>();
        for (Integer integer: cacheData){
            //这里对数据做一次深拷贝，避免篡改原始数据
            data.add(integer);
        }

        return data;
    }

}
